package com.ebs.md.api.abs;

/**
 * Perforce Revision:
 *  $File: //depot/MarketData/commons-md-api/MAIN/src/main/java/com/ebs/md/api/abs/AbstractOrderBook.java $
 *  $Revision: #11 $  $Change: 114604 $  $DateTime: 2012/09/02 11:22:42 $  $Author: Arik.Sher $
 */

import java.util.ArrayList;
import java.util.Calendar;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.NavigableSet;
import java.util.SortedSet;
import java.util.TreeMap;
import java.util.TreeSet;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.log4j.Logger;

import com.ebs.md.api.BookSide;
import com.ebs.md.api.IBookEntry;
import com.ebs.md.api.IOrder;
import com.ebs.md.api.IOrderBook;
import com.ebs.md.api.IOrderBookManager;
import com.ebs.md.api.IProduct;
import com.ebs.md.api.ISnapshot;
import com.ebs.md.api.OrderBookListener;
import com.ebs.md.api.OrderSide;
import com.ebs.md.api.Rate;
import com.ebs.md.api.comparators.MdComparators;
import com.ebs.md.api.orderbook.BookEntryPosition;
import com.ebs.md.api.orderbook.BookEntryPositionOfDeleted;


public abstract class AbstractOrderBook<BE extends IBookEntry, O extends IOrder> 
implements IOrderBook<BE>, IOrderBookManager<BE, O>
{
	// setting the Logger to private so will not be used directly. 
	//  instead, use the PrefixedLoggers.
    private   final Logger         basicLogger = Logger.getLogger(getClass());
    
    protected final PrefixedLogger   genLogger; // general logger(no specific side)
    protected final PrefixedLogger   bidLogger;
    protected final PrefixedLogger offerLogger;

	protected final static int MAX_MILLIS_BETWEEN_SNAPSHOTS = 3000;

	protected IProduct      product;
	protected NavigableMap<Rate, NavigableSet<BE>> bids;
	protected NavigableMap<Rate, NavigableSet<BE>> offers;
	
    protected OrderBookListener<IOrderBook<BE>, BE, O> orderBookListener;

	final protected Lock bidLock   = new ReentrantLock();
	final protected Lock offerLock = new ReentrantLock();
	
	protected AtomicBoolean isInitialized = new AtomicBoolean(false);
	
	// the following 2 members are used to collect snapshots(two sides)
	protected ISnapshot<BE> previousSideSnapshot;
	protected AtomicBoolean waitingForTheOtherSide = new AtomicBoolean(false);

	
	/**
	 * use this constructor for Empty book.
	 * 
	 * @param product
	 * @param readyForUse 
	 *         - when true,  could handle "applyOrder"
	 *         - when false, it's state is stale, at this state cannot handle "applyOrder".
	 *           only after snapshot(s) arrive than could handle apply order.
	 *           
	 */
	protected AbstractOrderBook(IProduct product, boolean readyForUse) {
		super();
		this.product    = product;
		  genLogger = new PrefixedLogger(basicLogger, product.getName() + ":");
		  bidLogger = new PrefixedLogger(basicLogger, product.getName() + "(BID):");
		offerLogger = new PrefixedLogger(basicLogger, product.getName() + "(OFFER):");
		initBidsWith  (Collections.EMPTY_LIST);
		initOffersWith(Collections.EMPTY_LIST);
		isInitialized.set(readyForUse);
	}

	/**
	 * use this constructor for Empty book, (uninitialized)
	 * 
	 * @param product
	 */
	protected AbstractOrderBook(IProduct product) {
		this(product, false);
	}


	/**
	 * use this constructor when BookEntries are ready and known.
	 * in this state the book can handle "applyOrder"
	 * 
	 * @param product
	 * @param bids
	 * @param offers
	 */
	protected AbstractOrderBook(IProduct product, List<BE> bids, List<BE> offers) {
		super();
		this.product    = product;
		  genLogger = new PrefixedLogger(basicLogger, product.getName() + ":");
		  bidLogger = new PrefixedLogger(basicLogger, product.getName() + "(BID):");
		offerLogger = new PrefixedLogger(basicLogger, product.getName() + "(OFFER):");
		initBidsWith  (bids);
		initOffersWith(offers);
		isInitialized.set(true);
	}

	public IProduct getProduct() {
		return product;
	}

	public void setOrderBookListener(OrderBookListener<IOrderBook<BE>, BE, O> listener) {
		this.orderBookListener = listener;
	}

	public void registerListener(OrderBookListener<IOrderBook<BE>, BE, O> listener) {
		setOrderBookListener(listener);
	}

	public List<BE> getAllBids() {
		List<BE> retList = new ArrayList<BE>();
		for ( Collection<BE> curr : bids.values() ) {
			retList.addAll(curr);
		}
		return retList;
	}

	public List<BE> getAllOffers() {
		List<BE> retList = new ArrayList<BE>();
		for ( Collection<BE> curr : offers.values() ) {
			retList.addAll(curr);
		}
		return retList;
	}
	
	public List<Rate> getTopBidRates  (int maxItems) {
		return subCollection(bids.  keySet(), maxItems);
	}
	public List<Rate> getTopOfferRates(int maxItems) {
		return subCollection(offers.keySet(), maxItems);
	}
	public Rate getBestBid() {
		if ( !isEmpty(bids) ) {
			return    bids.firstKey();
		}
		return null;
	}
	public Rate getBestOffer() {
		if ( !isEmpty(offers) ) {
			return    offers.firstKey();
		}
		return null;
	}

	public Rate getWorstBid() {
		if ( getBidRatesCount() > 1 ) {
			return bids.lastKey();
		}
		return null;
	}

	public Rate getWorstOffer() {
		if ( getOfferRatesCount() > 1 ) {
			return offers.lastKey();
		}
		return null;
	}

	@Override
	public Rate getNextBestRate(BookSide side, Rate rate) {
		NavigableMap<Rate, NavigableSet<BE>> bookAtSide = getAffectedOBSide(side.getOrderSide());

		return bookAtSide.higherKey(rate);
	}
	@Override
	public Rate getNextBestBid(Rate rate) {
		return getNextBestRate(BookSide.BID, rate);
	}
	@Override
	public Rate getNextBestOffer(Rate rate) {
		return getNextBestRate(BookSide.OFFER, rate);
	}

	public int getBidRatesCount() {
		return safeSize(bids);
	}

	public int getOfferRatesCount() {
		return safeSize(offers);
	}

	public int getBidEntriesCountAtRate(Rate rate) {
		if ( isEmpty(bids) ) {
			return 0;
		}
		SortedSet<BE>   entriesAtRateSet = bids.get(rate);
		return safeSize(entriesAtRateSet);
	}

	public int getOfferEntriesCountAtRate(Rate rate) {
		if ( isEmpty(offers) ) {
			return 0;
		}
		SortedSet<BE>   entriesAtRateSet = offers.get(rate);
		return safeSize(entriesAtRateSet);
	}

	public List<BE> getTopBidsAtRate  (Rate rate, int maxItems) {
		SortedSet<BE> bookEntriesSet = bids.  get(rate);
		return subCollection(bookEntriesSet, maxItems);
	}
	public List<BE> getTopOffersAtRate(Rate rate, int maxItems) {
		SortedSet<BE> bookEntriesSet = offers.get(rate);
		return subCollection(bookEntriesSet, maxItems);
	}

	protected NavigableMap<Rate, NavigableSet<BE>> allocSortedMap(Comparator<Rate> comp) {
		return new TreeMap<Rate, NavigableSet<BE>>(comp);
	}
	
	protected PrefixedLogger getLogger(BookSide bookSide) {
		if ( BookSide.BID.equals(bookSide) ) {
			return    bidLogger;
		}
		if ( BookSide.OFFER.equals(bookSide) ) {
			return    offerLogger;
		}
		return genLogger;
	}
	
	protected Lock getLockObject(BookSide bookSide) {
		if ( BookSide.BID.equals(bookSide) ) {
			return    bidLock;
		}
		if ( BookSide.OFFER.equals(bookSide) ) {
			return    offerLock;
		}
		return null; // not likely to get here
	}
	
	public void resetWithSnapshot(ISnapshot<BE> bidSideSnapshot, ISnapshot<BE> offerSideSnapshot) {
		// TODO: think of checking that product is the same...
		
		final boolean wasInitialized  = isInitialized.get();
		final boolean isFirstSnapshot = !wasInitialized;
		
		if (orderBookListener != null ) {
			orderBookListener.beforeSnapshot(this, bidSideSnapshot.getTime(), isFirstSnapshot);
		}
		initBidsWith  (  bidSideSnapshot.getEntries());
		initOffersWith(offerSideSnapshot.getEntries());
		if (!wasInitialized) {
			isInitialized.set(true);
		}
		postFullSnapshot();
		logFullSnapshotArrival(isFirstSnapshot, bidSideSnapshot.getTime());
		
		if (orderBookListener != null ) {
			orderBookListener. afterSnapshot(this, bidSideSnapshot.getTime(), isFirstSnapshot);
		}
	}

	protected void logFullSnapshotArrival(boolean firstTime, Calendar snapshotTime) {
		String extraLog = "";
		if ( firstTime ) {
			extraLog = " first time";
		}
		Date     snapshotDate = snapshotTime.getTime();
		genLogger.info(String.format("snapshot arrived (%s).%s", snapshotDate.toString(), extraLog));
	}
	
	/**
	 * 
	 * TODO: test this method...
	 */
	public void resetSnapshotSide(ISnapshot<BE> sideSnapshot) {
		if ( !this.product.equals(sideSnapshot.getProduct()) ) {
			// TODO: log it's not the same product...
			return;
		}
		
		if ( !waitingForTheOtherSide.get() ) {
			previousSideSnapshot = sideSnapshot;
			waitingForTheOtherSide.set(true);
			return;
		}
		
		if ( previousSideSnapshot.getSide().equals(sideSnapshot.getSide()) ) {
			previousSideSnapshot = sideSnapshot;
		} else {
			long millisDiff = Math.abs(sideSnapshot.getTime().getTimeInMillis() - previousSideSnapshot.getTime().getTimeInMillis());
			if ( millisDiff > MAX_MILLIS_BETWEEN_SNAPSHOTS ) {
				// irrelevant, too much time has passed...
				waitingForTheOtherSide.set(false);
				return;
			}
			// not same side + timeframe matches. both side are ready.
			ISnapshot<BE>   bidSideSnapshot = sideSnapshot;
			ISnapshot<BE> offerSideSnapshot = previousSideSnapshot;
			
			if ( BookSide.BID.equals(previousSideSnapshot.getSide()) ) {
				  bidSideSnapshot  = previousSideSnapshot;
				offerSideSnapshot  = sideSnapshot;
			} else {
				bidSideSnapshot    = sideSnapshot;
				offerSideSnapshot  = previousSideSnapshot;
			}
			
			resetWithSnapshot(bidSideSnapshot, offerSideSnapshot);
			
			waitingForTheOtherSide.set(false);
			previousSideSnapshot = null;
		}
	}

	protected void initBidsWith(Collection<BE> coll) {
		bidLock.lock();
		bids = allocSortedMap(MdComparators.bidRateComperator);
		addAll(bids  , coll);
		bidLock.unlock();
	}
	
	protected void initOffersWith(Collection<BE> coll) {
		offerLock.lock();
		offers = allocSortedMap(MdComparators.offerRateComperator);
		addAll(offers, coll);
		offerLock.unlock();
	}

	protected void addAll(NavigableMap<Rate, NavigableSet<BE>> dest, Collection<BE> src) {
		if ( src == null ) { 
			return;
		}
		for ( BE curr : src ) {
			
			SortedSet<BE> list = dest.get(curr.getRate());
			
			if ( list != null && !list.isEmpty() ) {
				list.add(curr);
			} else {
				dest.put(curr.getRate(), createBookEntriesAtRate(curr));
			}
		}
	}
	
	public void applyOrder(O order) {
		if ( !isInitialized.get() )  {
			PrefixedLogger logger = getLogger(BookSide.fromOrderSide(order.getSide()) );
			logger.warn(String.format("OrderBook is not initialized. cannot apply order with rate '%s'", order.getRate()));
			return;
			// TODO: think of throwing exception
//			throw new IllegalStateException("order book is not initialized, thus cannot apply orders...");
		}
		
		switch (order.getOperation()) {
		case ADD:
			handleAddition(order);
			break;
		case REMOVE:
		case CANCEL:
			handleRemoval(order);
			break;
		case INCREASE_AMOUNT:
			handleAmountIncrease(order);
			break;
		case DECREASE_AMOUNT:
			handleAmountDecrease(order);
			break;
		}
	}
	
	protected BookEntryPositionOfDeleted snapshotEntryPositionOfDeleted(BE be) {
		final BookSide bookSide = be.getSide();
		final Rate entryRate = be.getRate();
		NavigableMap<Rate, NavigableSet<BE>> bookAtSide = getAffectedOBSide(bookSide.getOrderSide());
		
		final int ratePosition = getRatePosition(bookAtSide, entryRate);
		if ( ratePosition <= 0 ) {
			getLogger(bookSide).warn(String.format("not probable but rate '%s' should have been found, since the BE was entered", entryRate.toString()));
			return null;
		}
		final int entriesAtPosition = getEnriesCountAtRate(bookAtSide, entryRate);
		final int numberOfPositions = bookAtSide.size();
		return new BookEntryPositionOfDeleted(ratePosition, entriesAtPosition, numberOfPositions);
	}
	
	protected BookEntryPosition snapshotEntryPosition(BE be) {
		final BookSide bookSide = be.getSide();
		final Rate entryRate = be.getRate();
		NavigableMap<Rate, NavigableSet<BE>> bookAtSide = getAffectedOBSide(bookSide.getOrderSide());

		final int ratePosition = getRatePosition(bookAtSide, entryRate);
		if ( ratePosition <= 0 ) {
			getLogger(bookSide).warn(String.format("not probable but rate '%s' should have been found, since the BE was entered", entryRate.toString()));
			return null;
		}
		final int entriesAtPosition = getEnriesCountAtRate(bookAtSide, entryRate);
		final int numberOfPositions = bookAtSide.size();
		return new BookEntryPosition(ratePosition, entriesAtPosition, numberOfPositions);
	}

	protected boolean isRateWithinRange(NavigableMap<Rate, NavigableSet<BE>> bookAtSide, Rate rate) {
		Comparator<? super Rate> bookSideCompr = bookAtSide.comparator();
		
		if ( rate != null
		     && bookSideCompr.compare(rate, bookAtSide.firstKey()) >= 0 
			 &&	bookSideCompr.compare(rate, bookAtSide.lastKey() ) <= 0
				)
		{
			return true;
		}
		return false;
	}
	
	/**
	 * 
	 * @param bookAtSide
	 * @param rate
	 * @return 1 if at top. 2 is second and so.
	 *         -1 means not found
	 */
	protected int getRatePosition(NavigableMap<Rate, NavigableSet<BE>> bookAtSide, Rate rate) {
		boolean found = false;
		int position = 0;
		
		if ( isRateWithinRange(bookAtSide, rate) ) {
			for ( Rate curr : bookAtSide.keySet() ) {
				position++;
				if ( curr.equals(rate) ) {
					found = true;
					break; // stop the loop
				}
			}
		}
		
		if ( found ) {
			return position;
		}
		return -1;
	}
	
	protected int getEnriesCountAtRate(NavigableMap<Rate, NavigableSet<BE>> bookAtSide, Rate rate) {
		NavigableSet<BE> sortedSet = bookAtSide.get(rate);
		
		int val = 0;
		
		if ( sortedSet != null ) {
			val = sortedSet.size();
		}
		return val;
	}

	protected NavigableMap<Rate, NavigableSet<BE>> getAffectedOBSide(OrderSide orderSide) {
		BookSide bookSide = BookSide.fromOrderSide(orderSide);
		
		if ( BookSide.BID.equals(bookSide) ) {
			return bids;
		}
		if ( BookSide.OFFER.equals(bookSide) ) {
			return offers;
		}
		throw new RuntimeException("book side not supported " + bookSide);
	}
	
	protected NavigableSet<BE> createBookEntriesAtRate(BE firstEntry) {
		NavigableSet<BE> retSet = new TreeSet<BE>(getBookEntryComparator());
		retSet.add(firstEntry);
		return retSet;
	}
	
	// taken from CollUtils
	public static <T> List<T> subCollection(Collection<T> coll, int maxItems) {
		if ( isEmpty(coll) ) {
			return Collections.emptyList();
		}
		if ( maxItems <= 0 ) {
			return Collections.emptyList();
		}
		
		int realMaxItems = Math.min(coll.size(), maxItems);
		
		List<T> retList = new ArrayList<T>(realMaxItems);
		
		Iterator<T> iterator = coll.iterator();
		for ( int ix=0 ; ix < realMaxItems && iterator.hasNext() ; ix++ ) {
			retList.add(iterator.next());
		}
		
		return retList;
	}
	
	// taken from CollUtils
	public static boolean isEmpty(Collection<?> coll) {
		return ( coll == null || coll.isEmpty() );
	}
	public static boolean isEmpty(Map<?, ?> map) {
		return ( map == null || map.isEmpty() );
	}
	public static int safeSize(Collection<?> coll) {
		if ( coll == null ) return 0;
		return coll.size();
	}
	public static int safeSize(Map<?, ?> map) {
		if ( map == null ) return 0;
		return map.size();
	}

	protected abstract void handleAddition(O order);
	protected abstract void handleRemoval(O order);
	protected abstract void handleAmountIncrease(O order);
	protected abstract void handleAmountDecrease(O order);
	
	protected abstract Comparator<BE> getBookEntryComparator();
	// postFullSnapshot - allow extending classes do extra stuff
	protected abstract void postFullSnapshot();
	
}
